package com.simple.rpc.component.server;

import com.simple.rpc.component.SimpleRpcComponent;
import com.simple.rpc.config.ApplicationConfig;
import com.simple.rpc.core.RpcServiceInvoker;
import com.simple.rpc.core.SimpleRpcServiceInvoker;
import com.simple.rpc.model.ServiceInvokeMetaData;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * socket方式打开服务端
 *
 * @author mhh
 * @since 2020/10/21
 */
@SimpleRpcComponent(name = {"default", "socket"})
public class SocketProviderServer implements ProviderServer {

    /**
     * 执行线程池
     */
    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(100, 100, 10, TimeUnit.MINUTES, new LinkedBlockingQueue<>(100));

    /**
     * 是否持续监听
     */
    private AtomicBoolean keepListening;

    /**
     * 开始监听
     */
    public void doListen(ApplicationConfig applicationConfig) throws IOException, ClassNotFoundException {
        ServerSocket serverSocket = new ServerSocket(Integer.parseInt(applicationConfig.getPort()));
        keepListening = new AtomicBoolean(true);
        while (keepListening.get()) {
            Socket accept = serverSocket.accept();
            System.out.println("接收到消息");
            InputStream inputStream = accept.getInputStream();
            ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
            Object readObject = objectInputStream.readObject();
            if (readObject instanceof ServiceInvokeMetaData) {
                //当监听到一个消费者请求 放到线程池里面继续操作
                threadPoolExecutor.submit(() -> {
                    RpcServiceInvoker rpcServiceInvoker = new SimpleRpcServiceInvoker();
                    Object result = rpcServiceInvoker.invoke((ServiceInvokeMetaData) readObject);
                    try {
                        OutputStream outputStream = accept.getOutputStream();
                        ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);
                        objectOutputStream.writeObject(result);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            }
        }
        threadPoolExecutor.shutdown();
    }

    @Override
    public void listen(ApplicationConfig applicationConfig) {
        if (StringUtils.isEmpty(applicationConfig.getHost()) || StringUtils.isEmpty(applicationConfig.getPort())) {
            System.out.println("applicationConfig 参数为空");
            throw new RuntimeException();
        }
        try {
            doListen(applicationConfig);
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
            throw new RuntimeException();
        }
    }

    @Override
    public void destroy() {
        keepListening.set(false);
    }
}
